package com.bingxu.flink.advancedcode;

import com.bingxu.flink.bean.WaterSensor;
import com.bingxu.flink.utils.TransFormat;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.GlobalWindows;
import org.apache.flink.streaming.api.windowing.assigners.ProcessingTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * @author :BingXu
 * @description :TODO
 * @date :2021/8/16 19:10
 * @modifier :
 */

public class GlobalWind {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.setInteger("port", 20000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);

        env
                .socketTextStream("localhost", 9999)
                .map((MapFunction<String, WaterSensor>) value -> {
                    String[] fields = value.split(",");
                    return new WaterSensor(fields[0], Long.valueOf(fields[1]), Integer.valueOf(fields[2]));
                })
                .keyBy(ele -> ele.getId())
                .window(GlobalWindows.create())
                .trigger(new Trigger<WaterSensor, GlobalWindow>() {

                    private Long cnt=0l;
                    // todo: 处理每一个元素的方法
                    /**
                     *  当有一个元素进入的时候统计个数
                     * @param element 传入元素
                     * @param timestamp 时间戳
                     * @param window 窗口
                     * @param ctx 上下文
                     * @return 返回触发器的结果
                     * @throws Exception
                     */
                    @Override
                    public TriggerResult onElement(WaterSensor element, long timestamp, GlobalWindow window, TriggerContext ctx) throws Exception {
                        cnt++;
                        if (cnt == 3) {
                            cnt = 0l;
                            return TriggerResult.FIRE_AND_PURGE;
                        }

                        return TriggerResult.CONTINUE;
                    }
                    //基于处理时间的处理方法
                    @Override
                    public TriggerResult onProcessingTime(long time, GlobalWindow window, TriggerContext ctx) throws Exception {
                        return null;
                    }

                    // 基于事件时间的处理方式
                    @Override
                    public TriggerResult onEventTime(long time, GlobalWindow window, TriggerContext ctx) throws Exception {
                        return null;
                    }

                    // 当触发器触发之后，清空窗口内的元素
                    @Override
                    public void clear(GlobalWindow window, TriggerContext ctx) throws Exception {
                        cnt = 0l;
                    }
                })
                .process(new ProcessWindowFunction<WaterSensor, String, String, GlobalWindow>() {
                    @Override
                    public void process(String s, Context context, Iterable<WaterSensor> elements, Collector<String> out) throws Exception {

                    }
                })
                .print();

        env.execute();
    }
}
